/*
 * Copyright 2015 The Netty Project
 *
 * The Netty Project licenses this file to you under the Apache License,
 * version 2.0 (the "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at:
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */
package io.netty.util.concurrent;

import io.netty.util.internal.DefaultPriorityQueue;
import io.netty.util.internal.ObjectUtil;
import io.netty.util.internal.PriorityQueue;

import java.util.Comparator;
import java.util.Queue;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * Abstract base class for {@link EventExecutor}s that want to support scheduling.
 */
public abstract class AbstractScheduledEventExecutor extends AbstractEventExecutor {

  private static final Comparator<ScheduledFutureTask<?>> SCHEDULED_FUTURE_TASK_COMPARATOR =
      new Comparator<ScheduledFutureTask<?>>() {
        @Override
        public int compare(ScheduledFutureTask<?> o1, ScheduledFutureTask<?> o2) {
          return o1.compareTo(o2);
        }
      };

  PriorityQueue<ScheduledFutureTask<?>> scheduledTaskQueue;

  protected AbstractScheduledEventExecutor() {
  }

  protected AbstractScheduledEventExecutor(EventExecutorGroup parent) {
    super(parent);
  }

  protected static long nanoTime() {
    return ScheduledFutureTask.nanoTime();
  }

  PriorityQueue<ScheduledFutureTask<?>> scheduledTaskQueue() {
    if (scheduledTaskQueue == null) {
      scheduledTaskQueue = new DefaultPriorityQueue<ScheduledFutureTask<?>>(
          SCHEDULED_FUTURE_TASK_COMPARATOR,
          // Use same initial capacity as java.util.PriorityQueue
          11);
    }
    return scheduledTaskQueue;
  }

  private static boolean isNullOrEmpty(Queue<ScheduledFutureTask<?>> queue) {
    return queue == null || queue.isEmpty();
  }

  /**
   * Cancel all scheduled tasks.
   *
   * This method MUST be called only when {@link #inEventLoop()} is {@code true}.
   */
  protected void cancelScheduledTasks() {
    assert inEventLoop();
    PriorityQueue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;
    if (isNullOrEmpty(scheduledTaskQueue)) {
      return;
    }

    final ScheduledFutureTask<?>[] scheduledTasks =
        scheduledTaskQueue.toArray(new ScheduledFutureTask<?>[0]);

    for (ScheduledFutureTask<?> task : scheduledTasks) {
      task.cancelWithoutRemove(false);
    }

    scheduledTaskQueue.clearIgnoringIndexes();
  }

  /**
   * @see #pollScheduledTask(long)
   */
  protected final Runnable pollScheduledTask() {
    return pollScheduledTask(nanoTime());
  }

  /**
   * Return the {@link Runnable} which is ready to be executed with the given {@code nanoTime}. You
   * should use {@link #nanoTime()} to retrieve the correct {@code nanoTime}.
   */
  protected final Runnable pollScheduledTask(long nanoTime) {
    assert inEventLoop();

    Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;
    ScheduledFutureTask<?> scheduledTask =
        scheduledTaskQueue == null ? null : scheduledTaskQueue.peek();
    if (scheduledTask == null) {
      return null;
    }

    // 这里取到的scheduledTask的deadlineNanos()比nanoTime要小，说明当前时间已经过了deadline，因而
    // 该task就是需要被执行的，因而将其从scheduledTaskQueue中移除
    if (scheduledTask.deadlineNanos() <= nanoTime) {
      scheduledTaskQueue.remove();
      return scheduledTask;
    }
    return null;
  }

  /**
   * Return the nanoseconds when the next scheduled task is ready to be run or {@code -1} if no task
   * is scheduled.
   */
  protected final long nextScheduledTaskNano() {
    Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;
    ScheduledFutureTask<?> scheduledTask =
        scheduledTaskQueue == null ? null : scheduledTaskQueue.peek();
    if (scheduledTask == null) {
      return -1;
    }
    return Math.max(0, scheduledTask.deadlineNanos() - nanoTime());
  }

  final ScheduledFutureTask<?> peekScheduledTask() {
    Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;
    if (scheduledTaskQueue == null) {
      return null;
    }
    return scheduledTaskQueue.peek();
  }

  /**
   * Returns {@code true} if a scheduled task is ready for processing.
   */
  protected final boolean hasScheduledTasks() {
    Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;
    ScheduledFutureTask<?> scheduledTask =
        scheduledTaskQueue == null ? null : scheduledTaskQueue.peek();
    return scheduledTask != null && scheduledTask.deadlineNanos() <= nanoTime();
  }

  @Override
  public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
    ObjectUtil.checkNotNull(command, "command");
    ObjectUtil.checkNotNull(unit, "unit");
    if (delay < 0) {
      delay = 0;
    }
    validateScheduled0(delay, unit);

    return schedule(new ScheduledFutureTask<Void>(
        this, command, null, ScheduledFutureTask.deadlineNanos(unit.toNanos(delay))));
  }

  @Override
  public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
    ObjectUtil.checkNotNull(callable, "callable");
    ObjectUtil.checkNotNull(unit, "unit");
    if (delay < 0) {
      delay = 0;
    }
    validateScheduled0(delay, unit);

    return schedule(new ScheduledFutureTask<V>(
        this, callable, ScheduledFutureTask.deadlineNanos(unit.toNanos(delay))));
  }

  @Override
  public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period,
      TimeUnit unit) {
    ObjectUtil.checkNotNull(command, "command");
    ObjectUtil.checkNotNull(unit, "unit");
    if (initialDelay < 0) {
      throw new IllegalArgumentException(
          String.format("initialDelay: %d (expected: >= 0)", initialDelay));
    }
    if (period <= 0) {
      throw new IllegalArgumentException(
          String.format("period: %d (expected: > 0)", period));
    }
    validateScheduled0(initialDelay, unit);
    validateScheduled0(period, unit);

    return schedule(new ScheduledFutureTask<Void>(
        this, Executors.<Void>callable(command, null),
        ScheduledFutureTask.deadlineNanos(unit.toNanos(initialDelay)), unit.toNanos(period)));
  }

  @Override
  public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay,
      TimeUnit unit) {
    ObjectUtil.checkNotNull(command, "command");
    ObjectUtil.checkNotNull(unit, "unit");
    if (initialDelay < 0) {
      throw new IllegalArgumentException(
          String.format("initialDelay: %d (expected: >= 0)", initialDelay));
    }
    if (delay <= 0) {
      throw new IllegalArgumentException(
          String.format("delay: %d (expected: > 0)", delay));
    }

    validateScheduled0(initialDelay, unit);
    validateScheduled0(delay, unit);

    return schedule(new ScheduledFutureTask<Void>(
        this, Executors.<Void>callable(command, null),
        ScheduledFutureTask.deadlineNanos(unit.toNanos(initialDelay)), -unit.toNanos(delay)));
  }

  @SuppressWarnings("deprecation")
  private void validateScheduled0(long amount, TimeUnit unit) {
    validateScheduled(amount, unit);
  }

  /**
   * Sub-classes may override this to restrict the maximal amount of time someone can use to
   * schedule a task.
   *
   * @deprecated will be removed in the future.
   */
  @Deprecated
  protected void validateScheduled(long amount, TimeUnit unit) {
    // NOOP
  }

  <V> ScheduledFuture<V> schedule(final ScheduledFutureTask<V> task) {
    if (inEventLoop()) {
      scheduledTaskQueue().add(task);
    } else {
      execute(new Runnable() {
        @Override
        public void run() {
          scheduledTaskQueue().add(task);
        }
      });
    }

    return task;
  }

  final void removeScheduled(final ScheduledFutureTask<?> task) {
    if (inEventLoop()) {
      scheduledTaskQueue().removeTyped(task);
    } else {
      execute(new Runnable() {
        @Override
        public void run() {
          removeScheduled(task);
        }
      });
    }
  }
}
